package engine

import (
	"log"
)

type ConcurrentEngine struct {
	Scheduler        Scheduler
	WorkerCount      int
	ItemChan         chan Item
	RequestProcessor Processor
}

type Processor func(request Request) (ParserResult, error)

type Scheduler interface {
	ReadyNotifier
	Submit(Request)
	WorkerChan() chan Request
	Run()
}

type ReadyNotifier interface {
	WorkerReady(chan Request)
}

func (engine *ConcurrentEngine) Run(seeds ...Request) ParserResult {

	engine.Scheduler.Run() //启动并行任务调度器
	out := make(chan ParserResult)

	for _, seed := range seeds {
		//需要receiver准备好才能发送，但是receiver准备需要下行代码worker工作
		//无限循环等待处理：开goroutine,不阻塞主协程
		engine.Scheduler.Submit(seed)
	}

	for i := 0; i < engine.WorkerCount; i++ {
		engine.createWorker(engine.Scheduler.WorkerChan(), out, engine.Scheduler)
	}

	for {
		result := <-out
		for _, item := range result.Items {
			go func(item Item) { engine.ItemChan <- item }(item)
		}
		for _, request := range result.Requests {
			engine.Scheduler.Submit(request)
		}
	}
}

// 网络慢，使用异步并发执行
// 参数，返回值，执行函数分离
func (engine *ConcurrentEngine) createWorker(in chan Request, out chan ParserResult, ready ReadyNotifier) {
	go func() {
		for {
			//执行方法 接收参数容器发送出去等待分配参数
			//等待调度器分配
			ready.WorkerReady(in)
			request := <-in
			result, err := engine.RequestProcessor(request)
			if err != nil {
				log.Printf("worker runing Error: %s\n", err)
				continue
			}
			out <- result
		}

	}()
}
